package com.tencent.tbds.demo.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

import static org.apache.hadoop.hbase.util.Bytes.toBytes;

/**
 * <p>
 * usage:
 * java -cp  dev-demo-<version>.jar:/usr/hdp/2.2.0.0-2041/hbase/lib/* com.tencent.tbds.demo.hbase.HBaseDemo --zk-host <zookeeper host list> --table-name <table name>
 */
public class HBaseDemo {
    public static void main(String[] args) throws IOException {
        HBaseDemoOption option = new HBaseDemoOption(args);
        if (option.hasHelp()) {
            option.printHelp();
            return;
        }
        String tableName = option.getTableName();

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", option.getZkHost());
        conf.set("zookeeper.znode.parent", "/hbase-unsecure");

        Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin();

        TableName[] tableNames = admin.listTableNames();
        System.out.println("Tables in HBase:");
        for (TableName name : tableNames) {
            System.out.println(name.getNameAsString());
        }

        System.out.println("----------------------------------------");
        //命名空间需要admin管理员创建
        //获取命名空间
        // String nameSpace = tableName.split(":")[0];
        // NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
        // admin.createNamespace(namespaceDescriptor);

        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor family = new HColumnDescriptor(toBytes("cf"));
        hTableDescriptor.addFamily(family);
        if (admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("table[" + tableName + "] already exists!");
        } else {
            admin.createTable(hTableDescriptor);
            System.out.println("Create table [" + tableName + "] success!");
        }

        System.out.println("----------------------------------------");

        Table table = connection.getTable(TableName.valueOf(tableName));

        Put put = new Put("1".getBytes());//—个 Put 代表一行数据，行键为构造方法中传入的值
        byte[] cf = "cf".getBytes();
        byte[] nameColumn = "name".getBytes();
        byte[] ageColumn = "age".getBytes();
        byte[] genderColumn = "gender".getBytes();
        byte[] scoreMathColumn = "scoreMath".getBytes();
        put.addColumn(cf, nameColumn, "liulv".getBytes());//本行数据的第一列
        put.addColumn(cf, ageColumn, "20".getBytes());// 本行数据的第二列
        put.addColumn(cf, genderColumn, "male".getBytes());// 本行數据的第三列
        put.addColumn(cf, scoreMathColumn, "99".getBytes());// 本行数据的第四列
        table.put(put);

        System.out.println("----------------------------------------");

        Scan scan = new Scan();
        scan.setBatch(50);
        scan.setCaching(100);
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(100);
        for (Result result : results) {
            String rowKey = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue(cf, nameColumn));
            String age = Bytes.toString(result.getValue(cf, ageColumn));
            String gender = Bytes.toString(result.getValue(cf, genderColumn));
            String scoreMath = Bytes.toString(result.getValue(cf, scoreMathColumn));
            System.out.println("======row key=====>>" + rowKey);
            System.out.println("======name=====>>" + name);
            System.out.println("======age=====>>" + age);
            System.out.println("======gender=====>>" + gender);
            System.out.println("======scoreMath=====>>" + scoreMath);
        }
        System.out.println("----------------------------------------");
        scanner.close();
        table.close();
        connection.close();

    }
}
